/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.testjob.tera;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Random;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.nhncorp.neptune.client.Cell;
import com.nhncorp.neptune.client.DirectUploader;
import com.nhncorp.neptune.client.NTable;
import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.util.FileUtil;
import com.nhncorp.neptune.parallel.hadoop.AbstractTabletInputFormat;
import com.nhncorp.neptune.parallel.hadoop.NeptuneMapReduceUtil;
import com.nhncorp.neptune.parallel.hadoop.SimpleInputFormat;
import com.nhncorp.neptune.tablet.TableSchema;

public class TeraBatchUploadJob {
  public static final Log LOG = LogFactory.getLog(TeraBatchUploadJob.class.getName()); 
  
  public void runJob(String tableName, int numOfTablets, int dataLength, int totalGb, String neptuneHome) throws IOException {
    JobConf jobConf = new JobConf(TeraJob.class);
    String jarPath = NeptuneMapReduceUtil.initMapReduce(jobConf, neptuneHome);
    NConfiguration nconf = new NConfiguration();
    
    if(!NTable.existsTable(nconf, tableName)) {
      TableSchema tableInfo = new TableSchema(tableName, "Test", new String[]{"Col1"});
      NTable.createTable(nconf, tableInfo);
    }
    Path tempOutputPath = new Path("TeraBatchUploadJob" + System.currentTimeMillis());
    
    jobConf.setJobName("TeraBatchUploadJob" + "(" + new Date() + ")");
    
    long rowsPerTask = ((((long)totalGb) * 1024L * 1024L * 1024L)/((long)dataLength)) / (long)numOfTablets;
    
    jobConf.setInt("teraJob.dataLength", dataLength);
    jobConf.setLong("teraJob.rowsPerTask", rowsPerTask);
    
    jobConf.setLong("mapred.task.timeout", 30 * 60 * 1000);
    
    //<MAP>
    jobConf.setMapperClass(TeraBatchUploadMap.class);
    jobConf.setInputFormat(SimpleInputFormat.class);
    jobConf.set(AbstractTabletInputFormat.OUTPUT_TABLE, tableName);
    jobConf.setNumMapTasks(numOfTablets);
    jobConf.setMapSpeculativeExecution(false);
    jobConf.setMaxMapAttempts(0);
    //</MAP>

    //<REDUCE>
    jobConf.setReducerClass(TeraBatchUploadReduce.class);
//    jobConf.setOutputKeyClass(Text.class);
//    jobConf.setOutputValueClass(Text.class);    
    FileOutputFormat.setOutputPath(jobConf, tempOutputPath);
    jobConf.setNumReduceTasks(1);
    //</REDUCE>
	    
    try {
	    //Run Job
	    JobClient.runJob(jobConf);
    } finally {
    	NeptuneMapReduceUtil.clearMapReduce(jarPath);
    	//delete temp output path
	    FileSystem fs = FileSystem.get(jobConf);
	    FileUtil.delete(fs, tempOutputPath, true);
	    
    }
  }
  
  static class TeraBatchUploadReduce implements Reducer<WritableComparable, Writable, WritableComparable, Writable> {
    public void reduce(WritableComparable key, Iterator<Writable> values, 
        OutputCollector<WritableComparable, Writable> collector, Reporter reporter) throws IOException {
    }

    public void configure(JobConf jobConf) {
    }

    public void close() throws IOException {
    }
  }
  
  static class TeraBatchUploadMap implements Mapper<WritableComparable, Writable, WritableComparable, Writable> {
    NTable ntable;
    NConfiguration conf;
    IOException err;
    DirectUploader uploader;
    int taskNum;
    long rowsPerTask;
    byte[] data;
    public void map(WritableComparable key, Writable value, 
        OutputCollector<WritableComparable, Writable> collector, 
        Reporter reporter) throws IOException {
      LOG.info("Start Map");
      if(err != null) {
        throw err;
      }

      DecimalFormat df = new DecimalFormat("00000000000000000000");
      
      //입력할 Tablet 추가
      Row.Key endRowKey = new Row.Key(df.format((long)taskNum * rowsPerTask + rowsPerTask - 1L));
      ntable.addTablet(endRowKey);
      
      this.uploader = ntable.openDirectUploader(new String[]{"Col1"});
      for(long i = 0; i < rowsPerTask; i++) {
        Row.Key rowKey = new Row.Key(df.format((long)taskNum * rowsPerTask + i));
        Row row = new Row(rowKey);
        row.addCell("Col1", new Cell(Cell.Key.EMPTY_KEY, this.data));
        uploader.put(row);
        
        if(i % 1000 == 0) {
          reporter.progress();
        }
        
        if( i % 10000 == 0) {
          LOG.info("uploaded: " + i);
        }
      }
      
      LOG.info("End Map");
    }

    public void configure(JobConf jobConf) {
      String taskId = jobConf.get("mapred.task.id");
      String taskId2 = taskId.substring(taskId.length() - 8);
      this.rowsPerTask = jobConf.getLong("teraJob.rowsPerTask", 0L);
      this.taskNum = Integer.parseInt(taskId2.substring(0,6));
      
      Random rand = new Random();
      this.data = new byte [jobConf.getInt("teraJob.dataLength", 1000)];
      rand.nextBytes(data);
      
      try {
        this.conf = new NConfiguration();
        this.ntable = NTable.openTable(conf, jobConf.get(AbstractTabletInputFormat.OUTPUT_TABLE));
      } catch (IOException e) {
        this.err = e;
      }
    }

    public void close() throws IOException {
      LOG.info("Start Map close");
      if(this.uploader != null) {
        this.uploader.close();
      }
      LOG.info("End Map close");
    }
  }
}
